package com.yht.redbook.count.biz.consumer;

import com.github.phantomthief.collection.BufferTrigger;
import com.google.common.collect.Lists;
import com.yht.framework.common.utils.JsonUtils;
import com.yht.redbook.count.biz.constant.MQConstants;
import com.yht.redbook.count.biz.constant.RedisKeyConstants;
import com.yht.redbook.count.biz.enums.CollectUnCollectTypeEnum;
import com.yht.redbook.count.biz.model.dto.AggregationCountCollectUnCollectNoteDTO;
import com.yht.redbook.count.biz.model.dto.CountCollectUnCollectNoteDTO;
import jakarta.annotation.Resource;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.messaging.Message;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.stereotype.Component;

import java.time.Duration;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.stream.Collectors;

/**
 * @author: yht
 * @date: 2025/2/23 21:28
 * @description:
 **/
@RocketMQMessageListener(consumerGroup = "redbook_group_" + MQConstants.TOPIC_COUNT_NOTE_COLLECT,
        topic = MQConstants.TOPIC_COUNT_NOTE_COLLECT)
@Slf4j
@Component
public class CountNoteCollectConsumer implements RocketMQListener<String> {

    @Resource
    private RedisTemplate<String, Object> redisTemplate;

    @Resource
    private RocketMQTemplate rocketMQTemplate;

    private BufferTrigger<String> bufferTrigger = BufferTrigger.<String>batchBlocking()
            .bufferSize(50000)
            .batchSize(1000)
            .linger(Duration.ofSeconds(1))
            .setConsumerEx(this::consumeMessage)
            .build();


    private void consumeMessage(List<String> bodys) {
        log.info("==> 【笔记收藏数】聚合消息，size：{}", bodys.size());
        log.info("==> 【笔记收藏数】聚合消息，{}", JsonUtils.toJsonString(bodys));

        List<CountCollectUnCollectNoteDTO> countCollectUnCollectNoteDTOS = bodys.stream()
                .map(body -> JsonUtils.parseObject(body, CountCollectUnCollectNoteDTO.class)).toList();

        Map<Long, List<CountCollectUnCollectNoteDTO>> groupMap = countCollectUnCollectNoteDTOS.stream()
                .collect(Collectors.groupingBy(CountCollectUnCollectNoteDTO::getNoteId));

        // 按钮汇总数据，统计出最终的计数
        // key 为笔记 ID，value 为最终操作的计数
        List<AggregationCountCollectUnCollectNoteDTO> countList = Lists.newArrayList();

        for (Map.Entry<Long, List<CountCollectUnCollectNoteDTO>> entry : groupMap.entrySet()) {
            Long noteId = entry.getKey();
            Long creatorId = null;

            List<CountCollectUnCollectNoteDTO> list = entry.getValue();

            int finalCount = 0;

            for (CountCollectUnCollectNoteDTO countCollectUnCollectNoteDTO : list) {
                Integer type = countCollectUnCollectNoteDTO.getType();
                creatorId = countCollectUnCollectNoteDTO.getNoteCreatorId();

                CollectUnCollectTypeEnum collectUnCollectTypeEnum = CollectUnCollectTypeEnum.valueOf(type);
                // 若枚举为空，则直接跳到下一循环
                if (Objects.isNull(collectUnCollectTypeEnum)) continue;

                switch (collectUnCollectTypeEnum) {
                    case COLLECT -> finalCount += 1;
                    case UNCOLLECT -> finalCount -= 1;
                }

            }
            // 将分组后统计的最终计数，存入 countMap 中
            countList.add(AggregationCountCollectUnCollectNoteDTO.builder()
                    .noteId(noteId)
                    .creatorId(creatorId)
                    .count(finalCount)
                    .build());
        }

        log.info("## 【笔记收藏数】聚合后的计数数据：{}", JsonUtils.toJsonString(countList));

        // 更新 Redis
        countList.forEach(item -> {
            // 笔记发布者 ID
            Long creatorId = item.getCreatorId();
            // 笔记 ID
            Long noteId = item.getNoteId();
            // 聚合后的计数
            Integer count = item.getCount();

            // 笔记 redis Key
            String countNoteRedisKey = RedisKeyConstants.buildCountNoteKey(noteId);
            // 判断 redis 中的笔记 hash 是否存在
            boolean isCountNoteExisted = redisTemplate.hasKey(countNoteRedisKey);

            // 如果 hash 存在，则更新 hash 中的收藏总数
            // 判断 hash 是否存在是初始化 hash 时会设置过期时间，防止 hash 过期，初始化 hash 的操作放在查询
            if (isCountNoteExisted) {
                redisTemplate.opsForHash().increment(countNoteRedisKey, RedisKeyConstants.FIELD_COLLECT_TOTAL, count);
            }

            // 更新 Redis 用户维度收藏数
            String countUserKey = RedisKeyConstants.buildCountUserKey(creatorId);
            boolean isCountUserExisted = redisTemplate.hasKey(countUserKey);
            if (isCountUserExisted) {
                redisTemplate.opsForHash().increment(countUserKey, RedisKeyConstants.FIELD_COLLECT_TOTAL, count);
            }
        });

        // 异步发送 MQ 将收藏总数落库
        Message<String> message = MessageBuilder.withPayload(JsonUtils.toJsonString(countList)).build();
        rocketMQTemplate.asyncSend(MQConstants.TOPIC_COUNT_NOTE_COLLECT_2_DB, message, new SendCallback() {
            @Override
            public void onSuccess(SendResult sendResult) {
                log.info("==> 【收藏总数落库】 MQ 发送成功，SendResult: {}", sendResult);
            }

            @Override
            public void onException(Throwable throwable) {
                log.info("==> 【收藏总数落库】 MQ 发送异常，", throwable);
            }
        });
    }

    @Override
    public void onMessage(String message) {
        // 往 bufferTrigger 中添加元素
        bufferTrigger.enqueue(message);
    }
}
